package vn.com.epi.recommender;

import java.io.IOException;
import java.util.Map;

import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import vn.com.epi.algorithm.clustering.Data;
import vn.com.epi.algorithm.knn.KNN;
import vn.com.epi.data.table.ArticleTable;
import vn.com.epi.data.table.UserTable;

import com.google.common.base.Strings;
import com.google.common.collect.Maps;

public class Trainer extends Configured implements Tool {
	
	private Map<String, Integer> userCode;
	private Map<Integer, Integer> category;
	private int[] categoryCode = {45,47,50,52,53,54,55,58,59,71,72,76,79,82,84,87,88,104,107,108,119,121,122,126,127,132,136,137,138,139,141,144,145,146,147,148,150,152};
	private int numberOfUsers;
	
	public Trainer() {
		category = Maps.newHashMap();
		userCode = Maps.newHashMap();
	}
	
	private int getNumBerOfUsers() throws IOException {
		int count = 0;
		HTable table = new HTable(UserTable.NAME);
		try {
            Scan scan = new Scan();
            ResultScanner rs = table.getScanner(scan);
            for (@SuppressWarnings("unused") Result r: rs) {
            	count++;
            }
		} finally {
			table.close();			
		}				
		return count;
	}
	
	private int getcategoryId(int categoryID) {
		int first = 0;
		int last = categoryCode.length - 1;
		while (first <= last) {
			int mid = (first + last)/2;
			if (categoryCode[mid] == categoryID)
				return mid + 1;
			else
				if (categoryCode[mid] < categoryID)
					first = mid + 1;					
				else
					last = mid - 1;					
		}
		int index = -1;
		return index;
	}		

	private void loadCategory() throws IOException {
        HTable table = new HTable(ArticleTable.NAME);
		try {
			Scan scan = new Scan();
			scan.addColumn(ArticleTable.FAM_MAIN, ArticleTable.COL_CATEGORY_ID);
			ResultScanner rs = table.getScanner(scan);
			for (Result r : rs) {
				int articleId = Bytes.toInt(r.getRow());
				int categoryId = Bytes.toInt(r.getValue(ArticleTable.FAM_MAIN, ArticleTable.COL_CATEGORY_ID));
				categoryId = getcategoryId(categoryId);
				if (categoryId < 0)
					continue;
				category.put(articleId, categoryId);
			}
		} finally {
			table.close();
		}
	}
	
	private void loadDataFromHbase() throws IOException {				
		HTable table = new HTable(UserTable.NAME);
		int index = 0;
		try {
			numberOfUsers = getNumBerOfUsers();
			new Data(numberOfUsers);
            Scan scan = new Scan();
            scan.addColumn(UserTable.FAM_MAIN, UserTable.COL_VISITED_ARTICLES);        
            ResultScanner rs = table.getScanner(scan);
            for (Result r: rs) {
            	String userId = Bytes.toString(r.getRow());
            	String line = UserTable.getVisitedArticles(r);
            	userCode.put(userId, index);
            	if (Strings.isNullOrEmpty(line)) {
    				index++;            	
            		continue;
            	}
            	Map<Integer, Integer> map = Maps.newHashMap();
            	String[] articles = line.split(",");
            	for (int i = 0; i < articles.length; i++) {
            		int articleId = Integer.parseInt(articles[i]);
					if (category.containsKey(articleId)) {
						int categoryId = category.get(articleId);
						if (map.containsKey(categoryId)) {
							int num = map.get(categoryId);
							num++;
							map.put(categoryId, num);
						} else
							map.put(categoryId, 1);
					}
				}
				for (int categoryId : map.keySet()) {
					int num = map.get(categoryId);
					Data.add(index, (short) categoryId, (short) num);
				}
				index++;            	
            }

		} finally {
			table.close();			
		}		
	}	
	
	@Override
	public int run(String[] arg0) throws Exception {
		loadCategory();
		loadDataFromHbase();
		int[] label = new int[numberOfUsers];
		Data.initial();
		KNN knn = new KNN(38);
		for (int i = 0; i < numberOfUsers; i++)
			label[i] = i;
		knn.optimize(numberOfUsers, label);
		knn.importNearestUser(userCode);
		return 0;
	}
	
	public static void main(String[] args) throws Exception {
        long startTime = System.currentTimeMillis();
        ToolRunner.run(new Trainer(), args);
        long endTime = System.currentTimeMillis();
        System.out.println("total time used for training: " + (endTime - startTime) + " ms.");
	}

}
